package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;

import com.atguigu.edu.realtime.common.base.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.bean.TradePaperStatistics;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DwsExaminePaperStatistics extends BaseApp {

    public static void main(String[] args) {
        new DwsExaminePaperStatistics().start(
                11111,
                4,
                "dws_exam_paper_exam_statistics_window",
                "dwd_exam_exam"
        );
    }
        @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );
       // jsonObjDS.print();

            SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                    WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                            .withTimestampAssigner(
                                    new SerializableTimestampAssigner<JSONObject>() {
                                        @Override
                                        public long extractTimestamp(JSONObject jsonObject, long l) {
                                            return jsonObject.getLong("ts") * 1000;
                                        }
                                    }
                            )
            );

            KeyedStream<JSONObject, String> paperIdKeyedDS = watermarkDS.keyBy(jsonObj -> jsonObj.getString("paper_id"));
//watermarkDS.print();
            SingleOutputStreamOperator<TradePaperStatistics> beanDS = paperIdKeyedDS.process(
                    new KeyedProcessFunction<String, JSONObject, TradePaperStatistics>() {
                        MapState<String, Long> userCount;
                        Long userSum = 0L;

                        @Override
                        public void open(Configuration parameters) throws Exception {
                            MapStateDescriptor<String, Long> userCt = new MapStateDescriptor<>("userCount", String.class, Long.class);
                            userCount = getRuntimeContext().getMapState(userCt);
                        }

                        @Override
                        public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradePaperStatistics>.Context context, Collector<TradePaperStatistics> collector) throws Exception {
                            String userId = jsonObject.getString("user_id");
                            if (userCount.get(userId) == null ) {
                                userCount.put(userId, 1L);
                                userSum = 1L;
                            }
                            String paperId = jsonObject.getString("paper_id");
                            String paperTitle = jsonObject.getString("paper_title");
                            Long durationSec = jsonObject.getLong("duration_sec");
                            Double source = jsonObject.getDouble("score");
                            Long ts = jsonObject.getLong("ts");
                            collector.collect(new TradePaperStatistics(
                                    "", "", "",
                                    paperId, paperTitle, userSum, source, durationSec, ts
                            ));
                        }
                    }
            );
            //beanDS.print();
            AllWindowedStream<TradePaperStatistics, TimeWindow> windowAllDS = beanDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

            SingleOutputStreamOperator<TradePaperStatistics> reduceDS = windowAllDS.reduce(
                    new ReduceFunction<TradePaperStatistics>() {
                        @Override
                        public TradePaperStatistics reduce(TradePaperStatistics v1, TradePaperStatistics v2) throws Exception {
                            v1.setTestExamPeople(v1.getTestExamPeople() + v2.getTestExamPeople());
                            v1.setTestAvgTime(v1.getTestAvgTime() + v2.getTestAvgTime());
                            v1.setTestAvgScore(v1.getTestAvgScore() + v2.getTestAvgScore());
                            return v1;
                        }
                    },
                    new AllWindowFunction<TradePaperStatistics, TradePaperStatistics, TimeWindow>() {
                        @Override
                        public void apply(TimeWindow timeWindow, Iterable<TradePaperStatistics> iterable, Collector<TradePaperStatistics> collector) throws Exception {
                            TradePaperStatistics orderBean = iterable.iterator().next();
                            orderBean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                            orderBean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                            orderBean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                            collector.collect(orderBean);
                        }
                    }
            );
            //reduceDS.print();
            SingleOutputStreamOperator<TradePaperStatistics> realDS = reduceDS.map(
                    new MapFunction<TradePaperStatistics, TradePaperStatistics>() {
                        @Override
                        public TradePaperStatistics map(TradePaperStatistics tradeClassStatisticsReal) throws Exception {
                            Long testAvgTime = tradeClassStatisticsReal.getTestAvgTime();
                            Double testAvg = tradeClassStatisticsReal.getTestAvgScore();
                            Long testExamPeople = tradeClassStatisticsReal.getTestExamPeople();

                            Long avgTime = testAvgTime / testExamPeople;
                            double avgScore = testAvg / testExamPeople;
                            return new TradePaperStatistics(
                                    tradeClassStatisticsReal.getStt(),
                                    tradeClassStatisticsReal.getEdt(),
                                    tradeClassStatisticsReal.getCurDate(),
                                    tradeClassStatisticsReal.getPaper_id(),
                                    tradeClassStatisticsReal.getPaper_title(),
                                    testExamPeople, avgScore, avgTime, tradeClassStatisticsReal.getTs()
                            );
                        }
                    }
            );
            realDS.print();
            realDS.map(new BeanToJsonStrMapFunction<>())
                    .sinkTo(FlinkSinkUtil.getDorisSink("dws_exam_paper_exam_statistics_window"));


        }
}
